Skip to main content

Testing

Testing apps is easy and enjoyable.

corva-sdk provides convenient tools for testing through pytest-plugin.

Write your tests using pytest to get the access to the plugin.

To install the library run:

pip install pytest
1. Stream​
1.1. Time​
from corva import Api, Cache, StreamTimeEvent, StreamTimeRecord, stream


@stream
def stream_app(event: StreamTimeEvent, api: Api, cache: Cache):
    return 'Hello, World!'


def test_stream_time_app(app_runner):
    event = StreamTimeEvent(
        asset_id=0, company_id=0, records=[StreamTimeRecord(timestamp=0)]
    )

    result = app_runner(stream_app, event=event)

    assert result == 'Hello, World!'
  • Sample app that we want to test.
  • Add app_runner argument to your test function.
  • Define the event that will be passed to the app.
  • Use app_runner fixture to run the app.
  • Verify the result.
1.2. Depth​
from corva import Api, Cache, StreamDepthEvent, StreamDepthRecord, stream


@stream
def stream_app(event: StreamDepthEvent, api: Api, cache: Cache):
    return 'Hello, World!'


def test_stream_depth_app(app_runner):
    event = StreamDepthEvent(
        asset_id=0, company_id=0, records=[StreamDepthRecord(measured_depth=0)]
    )

    result = app_runner(stream_app, event=event)

    assert result == 'Hello, World!'
  • Sample app that we want to test.
  • Add app_runner argument to your test function.
  • Define the event that will be passed to the app.
  • Use app_runner fixture to run the app.
  • Verify the result.
2. Scheduled​
2.1. Data Time​
from corva import Api, Cache, ScheduledDataTimeEvent, scheduled


@scheduled
def scheduled_app(event: ScheduledDataTimeEvent, api: Api, cache: Cache):
    return 'Hello, World!'


def test_scheduled_app(app_runner):
    event = ScheduledDataTimeEvent(
        asset_id=0, start_time=0, end_time=0, company_id=0
    )

    result = app_runner(scheduled_app, event=event)

    assert result == 'Hello, World!'
  • Sample app that we want to test.
  • Add app_runner argument to your test function.
  • Define the event that will be passed to the app.
  • Use app_runner fixture to run the app.
  • Verify the result.
2.2. Depth​

from corva import Api, Cache, ScheduledDepthEvent, scheduled


@scheduled
def scheduled_app(event: ScheduledDepthEvent, api: Api, cache: Cache):
    return 'Hello, World!'


def test_scheduled_app(app_runner):
    event = ScheduledDepthEvent(
        asset_id=0,
        company_id=0,
        top_depth=0.0,
        bottom_depth=1.0,
        log_identifier='',
        interval=1.0,
    )

    result = app_runner(scheduled_app, event=event)

    assert result == 'Hello, World!'
  • Sample app that we want to test.
  • Add app_runner argument to your test function.
  • Define the event that will be passed to the app.
  • Use app_runner fixture to run the app.
  • Verify the result.
2.3. Natural Time​
from corva import Api, Cache, ScheduledNaturalTimeEvent, scheduled


@scheduled
def scheduled_app(event: ScheduledNaturalTimeEvent, api: Api, cache: Cache):
    return 'Hello, World!'


def test_scheduled_app(app_runner):
    event = ScheduledNaturalTimeEvent(
        asset_id=0, company_id=0, schedule_start=0, interval=1
    )

    result = app_runner(scheduled_app, event=event)

    assert result == 'Hello, World!'
  • Sample app that we want to test.
  • Add app_runner argument to your test function.
  • Define the event that will be passed to the app.
  • Use app_runner fixture to run the app.
  • Verify the result.
3. Task​
from corva import Api, TaskEvent, task


@task
def task_app(event: TaskEvent, api: Api):
    return 'Hello, World!'


def test_task_app(app_runner):
    event = TaskEvent(asset_id=0, company_id=0)

    result = app_runner(task_app, event=event)

    assert result == 'Hello, World!'
  • Sample app that we want to test.
  • Add app_runner argument to your test function.
  • Define the event that will be passed to the app.
  • Use app_runner fixture to run the app.
  • Verify the result.